package com.atguigu.chapter05.Transform;

import com.atguigu.chapter05.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/3 15:29
 */
public class Flink09_SimpleRollAgg {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputDS = env.readTextFile("input/sensor.csv");

        SingleOutputStreamOperator<WaterSensor> sensorDS = inputDS.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                // 切分
                String[] line = value.split(",");
                return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));

            }
        });

        // TODO 简单的 滚动聚合算子
        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());
//        sensorKS.sum("vc").print("sum");
//        sensorKS.min("vc").print("min");
//        sensorKS.max("vc").print("max");

        sensorKS.maxBy("vc",true).print();
//        sensorKS.maxBy("vc",false).print();

        env.execute();
    }
}
/*
    聚合类的算子： 都在keyby之后使用， 因为都是 KeyedStream的算子，组内聚合

    min、max：除了比较字段之外，其他的字段取第一条的值
    minBy、maxBy：
        1）除了比较字段之外，其他的字段取最新的值
        2）如果比较的字段出现了相同值
            第二个参数为 true，其他字段 取 第一条的值
            第二个参数为 false，其他字段 取 最新的值
 */